package com.icehand.flink.demo;

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

public class Demo {
    public static void main(String[] args) throws Exception {
        final int port = 9008;
        final String host = "192.168.249.7";

        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStream<String> text = env.socketTextStream(host,port,"\n");
        DataStream<WordWithCount> wordCount = text.flatMap(
            new FlatMapFunction<String, WordWithCount>() {
                @Override
                public void flatMap(String s, Collector<WordWithCount> collector) throws Exception {
                    for(String word : s.split("\\s")) {
                        collector.collect(new WordWithCount(word,1L));
                    }
                }
            }
        )
        .keyBy("word")
        .timeWindow(Time.seconds(5),Time.seconds(1))
        .reduce(new ReduceFunction<WordWithCount>() {
            @Override
            public WordWithCount reduce(WordWithCount t1, WordWithCount t2) throws Exception {
                return new WordWithCount(t1.word, t1.count + t2.count);
            }
        });
        wordCount.print().setParallelism(1);
        env.execute("word count");
    }

    @Data
    @AllArgsConstructor
    @NoArgsConstructor
    public static class WordWithCount {
        private String word;
        private long count;
        @Override
        public String toString() {
            return word + " : " + count;
        }
    }
}
